package study.cdc.debezium.support.embedded;

import cn.hutool.json.JSONUtil;
import io.debezium.data.Envelope;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

@Slf4j
public class SourceRecordConsumer {

    public void handle(SourceRecord record) {
        log.info("", record.key());
        log.info("", record.keySchema());

        log.info("", record.value());
        log.info("", record.valueSchema());

        // key
        log.info("{}", record.sourceOffset());
        log.info("aa {}", JSONUtil.toJsonStr(record.key()));
        log.info("aa {}", JSONUtil.toJsonStr(record.keySchema()));
        // value
        Schema valueSchema = record.valueSchema();
        Struct payload = (Struct) record.value();

        String op = payload.getString(Envelope.FieldName.OPERATION);
        Envelope.Operation operation = Envelope.Operation.forCode(op);
        log.info(">>>>>> op= {}", op);
    }
}
